package com.lxn

import com.alibaba.fastjson.JSON
import com.alibaba.fastjson.serializer.SerializeConfig
import org.apache.http.HttpHost
import org.elasticsearch.action.bulk.BulkRequest
import org.elasticsearch.action.delete.DeleteRequest
import org.elasticsearch.action.get.{GetRequest, GetResponse}
import org.elasticsearch.action.index.IndexRequest
import org.elasticsearch.action.search.{SearchRequest, SearchRequestBuilder, SearchResponse}
import org.elasticsearch.action.update.UpdateRequest
import org.elasticsearch.client.{RequestOptions, RestClient, RestClientBuilder, RestHighLevelClient}
import org.elasticsearch.common.xcontent.XContentType
import org.elasticsearch.index.engine.Engine.{DeleteResult, GetResult}
import org.elasticsearch.index.query.{BoolQueryBuilder, QueryBuilders, TermQueryBuilder}
import org.elasticsearch.index.reindex.UpdateByQueryRequest
import org.elasticsearch.script.{Script, ScriptType}
import org.elasticsearch.search.aggregations.bucket.terms.{ParsedTerms, Terms}
import org.elasticsearch.search.aggregations.metrics.ParsedAvg
import org.elasticsearch.search.aggregations.{AggregationBuilders, BucketOrder}
import org.elasticsearch.search.{SearchHit, builder}
import org.elasticsearch.search.builder.SearchSourceBuilder
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder
import org.elasticsearch.search.sort.SortOrder

import java.util
import scala.collection.mutable.ListBuffer


object esTest {

  //编写scala的es代码，之前我学习的是java的es代码

  def main(args: Array[String]): Unit = {
//    println(client)
//    Put()
//    Post()
//    Bulk()
//    Update()
//    updateByQuery()
//    Delete()
//    GetById()
//    searchByQuery()
    searchByAggs()
    close(client)

  }

  //定义一个连接对象
  var client: RestHighLevelClient = create()

  //定义一个创建连接对象的方法
  def create(): RestHighLevelClient = {

    //创建配置对象
    val builder = RestClient.builder(new HttpHost("hadoop102", 9200))

    val client: RestHighLevelClient = new RestHighLevelClient(builder)
    client
  }

  //关闭连接对象
  def close(client: RestHighLevelClient): Unit = {
    if (client != null) client.close()
  }

  case class Movie(id:String,name:String){}

  //幂等插入数据
  def Put(): Unit = {
    //创建连接请求
    val request: IndexRequest = new IndexRequest()
    //连接的文档id
    request.id("101")
    //连接的索引
    request.index("movie_test1")
    //创建数据
    val movie : Movie = Movie("1001","速度与激情3")
    //将数据转换为json字符串  由于是样例类，需要序列化检测，防止循环引用报错
    val movieStr: String = JSON.toJSONString(movie,new SerializeConfig(true))
    //将数据放入请求体里面
    request.source(movieStr,XContentType.JSON)
    //发送请求  这种请求需要有请求参数，大多数情况下直接使用默认配置的参数即可
    client.index(request,RequestOptions.DEFAULT)
  }

  //非幂等插入操作
  def Post(): Unit = {
    //获取请求对象
    val request: IndexRequest = new IndexRequest()
    //指定索引  /非幂等操作不需要指定id
    request.index("movie_test1")
    //创建请求体数据
    val movie : Movie = Movie("102","超凡蜘蛛侠")
    //将数据放入请求体并进行序列化检验
    request.source(JSON.toJSONString(movie,new SerializeConfig(true)),XContentType.JSON)
    //发送非幂等请求
    client.index(request,RequestOptions.DEFAULT)
  }

  //批量操作，可以批量增删改查
  def Bulk(): Unit = {
    val bulkRequest: BulkRequest = new BulkRequest()

   /* //指定索引
    bulkRequest.add(new IndexRequest().index("movie_test1").id("1002").source(XContentType.JSON, "id", "103", "name", "寻梦环游记"))
    bulkRequest.add(new IndexRequest().index("movie_test1").id("1003").source(XContentType.JSON, "id", "104", "name", "星游记"))
    bulkRequest.add(new IndexRequest().index("movie_test1").id("1004").source(XContentType.JSON, "id", "105", "name", "神魄"))
    bulkRequest.add(new IndexRequest().index("movie_test1").id("1005").source(XContentType.JSON, "id", "106", "name", "一念永恒"))
    bulkRequest.add(new IndexRequest().index("movie_test1").id("1006").source(XContentType.JSON, "id", "107", "name", "魔道祖师"))*/

    //创建数据
    val movies: Array[Movie] = {
      Array[Movie](
        Movie("201", "水仙花"),
      Movie("202", "查理德堡"),
      Movie("203", "永生梦"))
    }
    //这样放在一个数组里面，集合里面也一样，都需要对其中每一个对象进行操作
    for(movie <- movies){
      //也可以不用new那么多对象
      val indexRequest: IndexRequest = new IndexRequest().index("movie_test1")
      //将数据放入请求体
      indexRequest.source(JSON.toJSONString(movie,new SerializeConfig(true)),XContentType.JSON)
      //指定id
      indexRequest.id(movie.id)
      //将插入响应添加至【批量处理
      bulkRequest.add(indexRequest)
    }
    val bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT)

    println(bulkResponse.getTook)
    println(bulkResponse.getItems)

  }

  //修改
  def Update(): Unit = {
    //获取修改请求对象
    val updateRequest: UpdateRequest = new UpdateRequest()
    //指定索引和id
    updateRequest.index("movie_test1").id("202")
    //修改，需要需要使用doc
    //其实在idea里面编写es代码和写es的dsl语句差不多，我们如果不会写，可以先将dsl语句写出来，在写代码
    //里面编写语句，前面写修改的字段，后面写修改的值
    updateRequest.doc(XContentType.JSON,"name","最终幻想2")
    //执行请求
    val updateResponse = client.update(updateRequest, RequestOptions.DEFAULT)
    println(updateResponse.getId)
  }


  //查询修改
  def updateByQuery(): Unit ={
    val updateByQueryRequest: UpdateByQueryRequest = new UpdateByQueryRequest("movie_index_cn")
    //query
    //val termQueryBuilder: TermQueryBuilder =
    //new TermQueryBuilder("movie_name.keyword","湄公河行动")
    val boolQueryBuilder: BoolQueryBuilder = QueryBuilders.boolQuery()
    val termQueryBuilder: TermQueryBuilder = QueryBuilders.termQuery("actorList.name.keyword", "张译")
    boolQueryBuilder.filter(termQueryBuilder)
    updateByQueryRequest.setQuery(boolQueryBuilder)
    //update
    val params: util.HashMap[String, AnyRef] = new util.HashMap[String,AnyRef]()
    params.put("newName" , "林惊羽")
    val script: Script = new Script(
      ScriptType.INLINE ,
      Script.DEFAULT_SCRIPT_LANG,
      "ctx._source['actorList']['name']=params.newName",
      params
    )
    updateByQueryRequest.setScript(script)

    client.updateByQuery(updateByQueryRequest , RequestOptions.DEFAULT)
  }

  //删除
  def Delete(): Unit = {
    //获取删除对象，指定索引和id
    val deleteRequest = new DeleteRequest("movie_test1", "skfa0pYBHCg2BrEvuLW1")
    client.delete(deleteRequest,RequestOptions.DEFAULT)
  }

  //根据id进行查询
  def GetById(): Unit = {
    //获取请求对象
    val getRequest: GetRequest = new GetRequest()
    //获取索引和id
    getRequest.index("movie_test1").id("1003")
    //执行请求
    val getResponse: GetResponse = client.get(getRequest, RequestOptions.DEFAULT)
    val result :String = getResponse.getSourceAsString
    println(result)

  }

  /**
   * search :
   *
   * 查询 doubanScore>=5.0 关键词搜索 red sea
   * 关键词高亮显示
   * 显示第一页，每页 20 条
   * 按 doubanScore 从大到小排序
   */

  def searchByQuery(): Unit = {
    //创建search的请求对象
    val searchRequest: SearchRequest = new SearchRequest()
    //查询条件需要放在source里面，里面封装一个builder对象
    //builder对象里面有dsl操作，如过滤，排序，高亮，分页等
    val searchSourceBuilder: SearchSourceBuilder = new SearchSourceBuilder()
    //然后使用这个builder里面的bool查询方法
    val boolQueryBuilder = QueryBuilders.boolQuery()
    //使用范围过滤
    val rangeQueryBuilder = QueryBuilders.rangeQuery("doubanScore").gte(5.0)
    val matchQueryBuilder = QueryBuilders.matchQuery("name", "red sea")
    boolQueryBuilder.filter(rangeQueryBuilder)
    boolQueryBuilder.must(matchQueryBuilder)
    searchSourceBuilder.sort("doubanScore",SortOrder.DESC)
    searchSourceBuilder.highlighter(new HighlightBuilder().field("name"))
    searchSourceBuilder.query(boolQueryBuilder)
    searchRequest.source(searchSourceBuilder)
    val response: SearchResponse = client.search(searchRequest, RequestOptions.DEFAULT)

    //这个其实和es的dsl语句的层级很像，searchSourceBuilder里面包含query,与他同级的有，排序，高亮，分页操作属于第一级操作
    //boolQueryBuilder是第二级操作，

    val hitsTotal : Long = response.getHits.getTotalHits.value
    val hits = response.getHits.getHits
    for( hit <- hits){
      val dataJson: String = hit.getSourceAsString
      val highLightValue: String = hit.getHighlightFields.get("name").fragments()(0).toString
      println(dataJson)
      println(highLightValue)
    }
  }

//  查询每位演员参演的电影的平均分，倒叙排序
def searchByAggs(): Unit = {
  //创建search语句的请求对象
  val searchRequest: SearchRequest = new SearchRequest()
  //开始编写聚合操作语句

  //使用聚合构建对象获取分组聚合对象
  val termsAggregationBuilder = AggregationBuilders.terms("groupByActor")
    .field("actorList.name.keyword").size(10)
    .order(BucketOrder.aggregation("doubanscoreavg",false))
  //avg操作
  val avgAggregationBuilder = AggregationBuilders.avg("doubanscoreavg").field("doubanScore")
  termsAggregationBuilder.subAggregation(avgAggregationBuilder)
  //创建一个编写语句对象
  val searchSourceBuilder: SearchSourceBuilder = new SearchSourceBuilder()
  searchSourceBuilder.aggregation(termsAggregationBuilder)
  //将编写语句放入source
  searchRequest.source(searchSourceBuilder)
  val response = client.search(searchRequest, RequestOptions.DEFAULT)

  val aggregations = response.getAggregations
  val aggTerms :ParsedTerms = aggregations.get[ParsedTerms]("groupByActor")
  val buckets : util.List[_ <: Terms.Bucket] = aggTerms.getBuckets
  import scala.collection.JavaConverters._
  for(bucket <- buckets.asScala){
    val name: String = bucket.getKeyAsString
    val doubanscoreavg: Double = bucket.getAggregations.get[ParsedAvg]("doubanscoreavg").getValue
    val count = bucket.getDocCount
    println(s"$name 共参演了 $count 部电影， 平均分为 $doubanscoreavg")

  }


  }
}
